package com._58city.spark.app;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

/**
 * kafka测试demo, 可以测试运行，打印kafka里的数据
 * @author zhaoxiang
 */
public class KafkaDemo {

	
	public static void custormerTest(){
		Properties props = new Properties();
		props.put("zookeeper.connect","10.126.99.105:2181,10.126.99.196:2181,10.126.81.208:2181,10.126.100.144:2181,10.126.81.215:2181/58_kafka_cluster");
		props.put("group.id","test"+new Random().nextInt(1000));
		props.put("zookeeper.session.timeout.ms", "400");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.offset.reset", "largest"); //largest
		props.put("auto.commit.interval.ms", "1000");
		//kafka topic
		String topic = "hdp_ubu_wuxian_58app_start";
		ConsumerConnector consumer=Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, 1);
		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
	    StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

	     Map<String, List<KafkaStream<String, String>>> consumerMap = 
	    		 consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
	     KafkaStream<String, String> stream = consumerMap.get(topic).get(0);
	     ConsumerIterator<String, String> it = stream.iterator();
	     long n = 1;
	     while (it.hasNext()){
	    	 	MessageAndMetadata<String, String> meta=it.next();
	    	 	System.out.println(meta.message());
	    	 	if(n == 2)
	    	 		break;
	    	 	n++;
	     }
	}
	
	
	public static void main(String[] args) {
		custormerTest();
		System.exit(0);
	}
}
